/*
 * Singularity Engine
 * Copyright (C) 2010 Vyacheslav Vorobyov, All rights reserved.
 * See LICENSE for full license information.
 */

/**
 * 
 */
package org.singularity.mapred.model;

import java.util.Iterator;
import java.util.Set;

import org.jgrapht.DirectedGraph;
import org.jgrapht.graph.DefaultDirectedGraph;
import org.jgrapht.traverse.TopologicalOrderIterator;

import org.singularity.io.Input;
import org.singularity.io.Output;
import org.singularity.mapred.Cascade;
import org.singularity.mapred.Destination;
import org.singularity.mapred.Element;
import org.singularity.mapred.InputElement;
import org.singularity.mapred.MapFun;
import org.singularity.mapred.OutputElement;
import org.singularity.mapred.RedFun;
import org.singularity.mapred.ReduceElement;
import org.singularity.mapred.Source;

/**
 * @author vjache
 *
 */
public class DefaultCascade implements Cascade {
	
//	private final HashSet<Element> _vertexes = new HashSet<Element>();
//	private final HashSet<Flow> _edges = new HashSet<Flow>();
	
	/**
	 * @return the graph
	 */
	public DirectedGraph<Element, Flow> getGraph() {
		return _graph;
	}

	private final DirectedGraph<Element, Flow> _graph = 
		new DefaultDirectedGraph<Element, Flow>(Flow.class);
	
	private int _nextId = 0;

	/**
	 * 
	 */
	public DefaultCascade() {
	}
	
	private int nextId()
	{
		return _nextId++;
	}

	@Override
	public <VT> InputElement<VT> createInput(Input<VT> aInput) {
		final DefaultInputElement<VT> outputElem = new DefaultInputElement<VT>(this,aInput,nextId());
		_graph.addVertex(outputElem);
		return outputElem;
	}

	@Override
	public <VT> OutputElement<VT> createOutput(Output<VT> aOutput) {
		final DefaultOutputElement<VT> outputElem = new DefaultOutputElement<VT>(this,aOutput,nextId());
		_graph.addVertex(outputElem);
		return outputElem;
	}

	@Override
	public <KT, RT> ReduceElement<KT, RT> createReducer(RedFun<KT, RT> aRedFun) {
		final DefaultReduceElement<KT, RT> redElem = new DefaultReduceElement<KT, RT>(this,aRedFun,nextId());
		_graph.addVertex(redElem);
		return redElem;
	}

	@Override
	public <VT> void flow(Source aFrom, OutputElement<VT> aTo) {
		assertVertexExists(aFrom);
		assertVertexExists(aTo);
		
		final OutputFlow<VT> outputFlow = new OutputFlow<VT>(nextId(),this, aFrom, aTo);
		_graph.addEdge(aFrom, aTo, outputFlow);
	}

	@Override
	public <VT0, KT, VT> void flow(Source aFrom, Destination aTo,
			MapFun<VT0, KT, VT> aMapFun) {
		assertVertexExists(aFrom);
		assertVertexExists(aTo);
		
		final MapFlow<VT0, KT, VT> mapFlow = new MapFlow<VT0, KT, VT>(nextId(),this, aFrom, aTo, aMapFun);
		_graph.addEdge(aFrom, aTo, mapFlow);
	}
	
	private void assertVertexExists(Element aElem)
	{
		if(!_graph.containsVertex(aElem))
			throw new RuntimeException(String.format("%s not existed in cascade %s.",aElem,this));
	}
	
	public Iterable<Element> topologyOrdered()
	{
		return new Iterable<Element>() {
			@Override
			public Iterator<Element> iterator() {
				return new Iterator<Element>() {
					private final TopologicalOrderIterator<Element, Flow> _iter = new TopologicalOrderIterator<Element, Flow>(_graph);
					@Override
					public void remove() {_iter.remove();}
					@Override
					public Element next() {return _iter.next();}
					@Override
					public boolean hasNext() {return _iter.hasNext();}
				};
			}
		};
	}
	
	public Set<Flow> outflows(Element aFrom)
	{
		return _graph.outgoingEdgesOf(aFrom);
	}
	
	public Set<Flow> inflows(Element aTo)
	{
		return _graph.incomingEdgesOf(aTo);
	}

	@Override
	public String toString() {
		return _graph.toString();
	}
}
